package cn.dmp.report

import java.util.Properties

import com.typesafe.config.ConfigFactory
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

object RptAreaAnalysis {

    def main(args: Array[String]): Unit = {

        // 检验参数 -> 配置文件
        if (args.length != 2) {
            println(
                """
                  |cn.dmp.report.RptAreaAnalysis
                  |参数：dataInputPath, outputPath
                """.stripMargin)
            sys.exit()
        }

        val Array(dataInputPath, outputPath) = args

        // sparkcontext
        val sparkConf = new SparkConf().setMaster("local[*]").setAppName("地域报表")
        val sc = new SparkContext(sparkConf)

        // 读取数据 - parquet
        val sqlc = new SQLContext(sc)
        // 数据文件 - dataframe
        val parquet = sqlc.read.parquet(dataInputPath)
        // 注册临时表
        parquet.registerTempTable("log")
        // 处理数据
        val resultSet = sqlc.sql(
            """
              |select provincename, cityname,
              |sum(case when requestmode=1 and processnode>=1 then 1 else 0 end) rawreq,
              |sum(case when requestmode=1 and processnode>=2 then 1 else 0 end) effreq,
              |sum(case when requestmode=1 and processnode=3 then 1 else 0 end) adreq,
              |sum(case when iseffective=1 and isbilling=1 and isbid=1 and adorderid !=0 then 1 else 0 end) rtbreq,
              |sum(case when iseffective=1 and isbilling=1 and iswin=1 then 1 else 0 end) rtbsucc,
              |sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) adshow,
              |sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) adclick,
              |sum(case when iseffective=1 and isbilling=1 and iswin=1 then winprice/1000 else 0 end) cb,
              |sum(case when iseffective=1 and isbilling=1 and iswin=1 then adpayment/1000 else 0 end) xf
              |from log
              |group by provincename, cityname
            """.stripMargin)


        /*val sql2 =
            """
              |select provincename,cityname,
              |sum(if(requestmode=1 and processnode>=1,1,0)) as request1,
              |sum(if(requestmode=1 and processnode>=2,1,0)) as request2,
              |sum(if(requestmode=1 and processnode=3,1,0)) as request3,
              |sum(if(iseffective=1 and isbilling=1 and isbid =1 and adorderid !=1,1,0)) as request4,
              |sum(if(iseffective=1 and isbilling=1 and iswin = 1,1,0)) as request5,
              |sum(if(requestmode=2 and iseffective=1,1,0)) as request6,
              |sum(if(requestmode=3 and iseffective=1,1,0)) as request7,
              |sum(if(iseffective =1 and isbilling=1 and iswin =1,winprice/1000,0)) as request8,
              |sum(if(iseffective =1 and isbilling=1 and iswin =1,adpayment/1000,0)) as request9
              |from log
              |group by provincename,cityname
            """.stripMargin*/

        // 存储数据
        val config = ConfigFactory.load()
        val props = new Properties()
        props.setProperty("user", config.getString("mysql.user"))
        props.setProperty("password", config.getString("mysql.password"))
        // 存储到数据库中
        resultSet.write.jdbc(
            config.getString("mysql.url"),
            "rptarea_28",
            props
        )

        sc.stop()

    }

}
